package demo;


import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * 1、Slot的特点
 * 1）均分内存，但是不隔离CPU
 * 2）可以共享，同一个job中，不同算子的子任务才可以共享，同一个slot，同时在运行的前提是，属于同一个slot共享组，默认都是default。
 * 2、Slot的数量和并行度的关系
 * 1）slot是一种静态的概念，表示最大的并发上线，并行度是一种动态的概念，表示实际运行线程数。
 * 2）slot数量>=job并行度（算子并行度），job才能运行
 *
 *
 *
 *
 * 比如我是8核16线程的电脑，我现在要设置3个TaskManager，那每个TaskManager要设置几个slot？
 *
 *电脑是 8核16线程 的配置，这意味着你的 CPU 有 8 个物理核心和 16 个逻辑线程（通过超线程技术实现）。你计划运行 3个 TaskManager，需要决定每个 TaskManager 的 slot 数量。
 *
 * 关键考虑因素
 * CPU 核心数与 Slot 数量的关系：
 * 每个 slot 可以看作是一个任务执行的单元，通常会占用一定的 CPU 资源。
 * 如果 slot 数量过多，可能会导致任务之间争夺 CPU 资源，增加上下文切换开销。
 * 如果 slot 数量过少，则可能无法充分利用 CPU 资源。
 * TaskManager 的数量与总资源分配：
 * 你的电脑有 16 个逻辑线程，假设这些线程可以完全被利用，那么总共可以支持大约 16 个并发任务（具体取决于任务的类型）。
 * 如果你设置了 3 个 TaskManager，那么需要将这 16 个逻辑线程合理分配给每个 TaskManager。
 * 任务类型的影响：
 * 计算密集型任务：建议 slot 数量接近或等于可用的 CPU 核心数（即每个 TaskManager 的 slot 数量应较小）。
 * I/O 密集型任务：可以设置更多的 slot 数量，因为 I/O 操作会让 CPU 空闲，允许更多任务并发运行。
 * 推荐配置
 * 假设你的电脑有 16 个逻辑线程，并且你要运行 3 个 TaskManager，以下是几种可能的配置方案：
 *
 * 方案 1：均匀分配
 * 每个 TaskManager 分配 5 或 6 个 slot。
 * 方案 2：偏向计算密集型任务
 * 每个 TaskManager 分配 4 或 5 个 slot。
 * 方案 3：偏向 I/O 密集型任务
 * 每个 TaskManager 分配 6 或 7 个 slot。
 *
 */
public class Flink07_YARN {
    public static void main(String[] args) throws Exception {

        /**
         * 创建执行环境
         * IDEA运行的时候，
         * 需要导入依赖
         */
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());


        /**
         * 设置全局并行度，全局并行度的优先级低于在算子上指定，默认的并行度和线程数有关，Socket这种只有一个并行度。
         * 也可以在WEB界面，提交JAR包的时候设置并行度。
         */
        executionEnvironment.setParallelism(3);


        /**
         * 全局禁用算子链
         */
        executionEnvironment.disableOperatorChaining();

        /**
         * 读取数据，此处为从文件中读取数据
         */

        DataStreamSource<String> stringDataStreamSource = executionEnvironment.socketTextStream("192.168.159.104",7777);

        /**
         * 进行切分，分组，聚合
         */
        SingleOutputStreamOperator<Tuple2<String, Integer>> streamOperator = stringDataStreamSource.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                String[] s1 = s.split(" ");

                for (String s2 : s1) {
                    Tuple2<String, Integer> tuple2 = new Tuple2<>(s2, 1);
                    collector.collect(tuple2);
                }
            }
        });

        /**
         * Java中泛型擦除的存在，在某些特殊情况下（比如Lambda表达式中），自动提取的信息是不够精细的——只告诉Flink当前的元素由“船头、船身、船尾”构成，
         * 根本无法重建出“大船”的模样；这时就需要显式地提供类型信息，才能使应用程序正常工作或提高其性能
         *
         * 如果用一下的写法，会报错，因为Flink不知道返回的Tuple中的字段的具体类型是什么,需要进行显式的指定
         */
        SingleOutputStreamOperator<Tuple2<String, Integer>> lamda = stringDataStreamSource.flatMap(
                (String s, Collector<Tuple2<String, Integer>> collector)->{
                    String[] s1 = s.split(" ");
                    for (String s2 : s1) {
                        Tuple2<String, Integer> tuple2 = new Tuple2<>(s2, 1);
                        collector.collect(tuple2);
                    }
                }
        ).returns(Types.TUPLE(Types.STRING,Types.INT));


        KeyedStream<Tuple2<String, Integer>, String> tuple2StringKeyedStream = streamOperator.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> stringIntegerTuple2) {
                return stringIntegerTuple2.f0;
            }
        });



        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = tuple2StringKeyedStream.sum(1);


        sum.print();

        executionEnvironment.execute();


    }
}
